package com.k2data.k2app.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Maps;
import com.k2data.k2app.common.MyConstant;
import com.k2data.k2app.domain.po.EarlyWaringDO;
import com.k2data.k2app.mapper.EarlyWaringMapper;
import com.k2data.k2app.service.DeviceService;
import com.k2data.k2app.service.RedisService;
import com.k2data.k2app.util.DateFormatterUtil;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.Decoder;
import kafka.utils.VerifiableProperties;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author lidong9144@163.com 17-7-13.
 */
@Log4j2
@Component
public class KafkaService {

    @Value("${kafka.consumer.num}")
    private Integer threadNum;
    @Value("${kafka.consumer.topic}")
    private String topic;
    @Value("${kafka.consumer.topicData}")
    private String topicData;
    @Value("${kafka.zookeeper.url}")
    private String zookeeperUrl;

    @Autowired
    @Lazy
    private DeviceService deviceService;
    @Autowired
    @Lazy
    private RedisService redisService;

    private ConsumerConnector consumer;
    private ExecutorService executorService;
    private final EarlyWaringMapper earlyWaringMapper;

    private static final Long expiresIn = 10 * 60L;
    private Cache<String, String> cache = CacheBuilder.newBuilder()
            .maximumSize(3000)
            .expireAfterWrite(expiresIn, TimeUnit.SECONDS)
            .build();

    @Autowired
    public KafkaService(EarlyWaringMapper earlyWaringMapper) {
        this.executorService = Executors.newCachedThreadPool();
        this.earlyWaringMapper = earlyWaringMapper;
    }

    public void init() {
        Properties props = new Properties();
        props.put("zookeeper.connect", this.zookeeperUrl);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "largest");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-k2asset");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        this.consumer = Consumer.createJavaConsumerConnector(new kafka.consumer.ConsumerConfig(props));
    }

    public void consumeTopic() {
        Map<String ,Integer> topicCountMap = new HashMap<>(16);
        topicCountMap.put(this.topic, this.threadNum);
        topicCountMap.put(this.topicData, this.threadNum);

        Decoder<String> keyDecoder = new kafka.serializer.StringDecoder(new VerifiableProperties());
        Decoder<String> valueDecoder = new kafka.serializer.StringDecoder(new VerifiableProperties());

        Map<String, List<KafkaStream<String, String>>> map = this.consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);

        List<KafkaStream<String, String>> kafkaStreams = map.get(this.topic);

        for (KafkaStream<String, String> kafkaStream : kafkaStreams) {
            this.executorService.submit(runnable(kafkaStream.iterator()));
        }

        List<KafkaStream<String, String>> kafkaStreamsData = map.get(this.topicData);

        for (KafkaStream<String, String> kafkaStream : kafkaStreamsData) {
            this.executorService.submit(runnableData(kafkaStream.iterator()));
        }
    }
    private Runnable runnable(ConsumerIterator<String, String> iterator) {
        return () -> {
            try {
                while (iterator.hasNext()) {
                    MessageAndMetadata<String, String> messageAndMetadata = iterator.next();

                    JSONObject object = JSON.parseObject(messageAndMetadata.message());

                    String deviceNo = object.getJSONObject("compoundId").getString("factory_id")
                            + '_'
                            + object.getJSONObject("compoundId").getString("equipment_id")
                            + '_'
                            + object.getJSONObject("compoundId").getString("machine_id");
                    Long timestamp = object.getLong("timestamp");

                    JSONArray results = object.getJSONArray("fieldResults");
                    for (Object result : results) {
                        JSONObject jsonObject = JSON.parseObject(result.toString());
                        Integer level = jsonObject.getInteger("level");
                        String message = jsonObject.getString("message");

                        EarlyWaringDO latest = selectLatest(deviceNo, message, level);

                        if (latest != null && latest.getWaringTime().plusMinutes(10).isAfter(LocalDateTime.now())) {
                            System.out.println("十分钟重复告警，不发送！");
                            continue;
                        }
                        EarlyWaringDO earlyWaringDO = new EarlyWaringDO();
                        earlyWaringDO.setDeviceNo(deviceNo);
                        earlyWaringDO.setLevel(MyConstant.getStatus(level));
                        earlyWaringDO.setMessage(message);
                        earlyWaringDO.setWaringTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()));
                        earlyWaringDO.setType("real");
                        earlyWaringDO.setFaultName("实时预警");
                        earlyWaringDO.setIsMq(1);
                        System.out.println("insert a warning! "+ earlyWaringDO.getMessage());
                        earlyWaringMapper.insert(earlyWaringDO);
                        deviceService.changeStatus(deviceNo, MyConstant.getStatus(level));
                        Map map = redisService.get(deviceNo);
                        map.put("status",MyConstant.getStatus(level));
                        redisService.set(deviceNo, map);
                    }
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        };
    }

    private Runnable runnableData(ConsumerIterator<String, String> iterator) {
        return () -> {
            try {
                while (iterator.hasNext()) {
                        MessageAndMetadata<String, String> messageAndMetadata = iterator.next();
                        JSONObject object = JSON.parseObject(messageAndMetadata.message());
                        JSONArray results = object.getJSONArray("fields");
                        JSONObject jsonObjectTime = object.getJSONObject("sampleTime");
                        Long timestamp = jsonObjectTime.getLong("timestamp");
                        String ts = "";
                        if(timestamp==null){
                            ts = parseTime(jsonObjectTime.getString("iso"));
                        }else{
                            LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault());
                            ts = DateFormatterUtil.parseToString(localDateTime);
                        }

                        Map<String,String> map = Maps.newHashMap();
                        for (Object result : results) {
                            JSONObject jsonObject = JSON.parseObject(result.toString());
                            if(jsonObject.getString("fieldId").equals("factory_id")||jsonObject.getString("fieldId").equals("equipment_id")
                                    ||jsonObject.getString("fieldId").equals("machine_id")
                                    ||jsonObject.getString("fieldId").equals("speed")){
                                map.put(jsonObject.getString("fieldId"),jsonObject.getString("fieldValue"));
                            }
                        }
                        String  deviceNo = new StringBuffer("").append(map.get("factory_id")).append("_")
                                .append(map.get("equipment_id")).append("_").append(map.get("machine_id")).toString();
                        String status = MyConstant.DEVICE_STATUS_HEALTH;
                        if(map.get("speed") != null){
                            if(Double.parseDouble(map.get("speed"))==0){
                                status = MyConstant.DEVICE_STATUS_STOP;
                            }
                        }
                        if(redisService.get(deviceNo) != null){
                            Object status1 = redisService.get(deviceNo).get("status");
                            if(status1 != null && status1.toString().equals(status)){
                            } else {
                                // 状态发生变化，更新数据库中设备状态
                                deviceService.changeStatus(deviceNo, status);
                                // 状态变为停机，发送停机告警
                                if(status.equals(MyConstant.DEVICE_STATUS_STOP)){
                                    if (cache.getIfPresent(deviceNo) == null) {
                                        cache.put(deviceNo, "1");
                                        EarlyWaringDO latest = earlyWaringMapper.selectLatestData(deviceNo, "设备转速为0，设备状态置为停机，请管理员及时检查", status);
                                        // 十分钟之内有重复告警，不发送
                                        if (latest != null && latest.getWaringTime().plusMinutes(10).isAfter(LocalDateTime.now())) {
                                            continue;
                                        }
                                        EarlyWaringDO earlyWaringDO = new EarlyWaringDO();
                                        earlyWaringDO.setDeviceNo(deviceNo);
                                        earlyWaringDO.setLevel(MyConstant.DEVICE_STATUS_STOP);
                                        earlyWaringDO.setMessage("设备转速为0，设备状态置为停机，请管理员及时检查");
                                        earlyWaringDO.setWaringTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()));
                                        earlyWaringDO.setType("real");
                                        earlyWaringDO.setFaultName("实时预警");
                                        earlyWaringDO.setIsMq(1);
                                        if(earlyWaringDO.getWaringTime().isAfter(LocalDateTime.now())){
                                        }else{
                                            earlyWaringMapper.insert(earlyWaringDO);
                                        }
                                    }
                                }
                            }
                        }
                        Map deviceInfo = Maps.newHashMap();
                        deviceInfo.put("ts",ts);
                        deviceInfo.put("status",status);
                        redisService.set(deviceNo,deviceInfo);
                    }

            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        };
    }

    public void shutdown() {
        this.executorService.shutdown();
    }

    private EarlyWaringDO selectLatest(String deviceNo, String message, Integer level) {
//        EarlyWaringDO earlyWaringDO = new EarlyWaringDO();
//        earlyWaringDO.setDeviceNo(deviceNo);
//        earlyWaringDO.setLevel(MyConstant.getStatus(level));
//        earlyWaringDO.setMessage(message);
        return earlyWaringMapper.selectLatestData(deviceNo, message, MyConstant.getStatus(level));
    }

    public static  String parseTime(String iso) {
//        String iso = "2017-02-01T00:03:39.510+00:00";
        LocalDateTime localDateTime = LocalDateTime.of(Integer.parseInt(iso.substring(0,4)),
                Integer.parseInt(iso.substring(5,7)),
                Integer.parseInt(iso.substring(8,10)),
                Integer.parseInt(iso.substring(11,13)),
                Integer.parseInt(iso.substring(14,16)),
                Integer.parseInt(iso.substring(17,19))).plusHours(8);
        return DateFormatterUtil.parseToString(localDateTime);
    }
}
